package org.penguin.study.bk.mqtt.integration;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.thymeleaf.util.DateUtils;

import java.util.*;
import java.util.stream.Collectors;

/**
 * @author chengyinghua
 */
@Configuration
@Slf4j
@ConditionalOnProperty(value = "mqtt.enabled", havingValue = "true")
public class MqttConfig {

    @Value("${mqtt.url}")
    private String url;

    @Value("${mqtt.topics}")
    private String topics;

    @Value("${mqtt.username}")
    private String username;

    @Value("${mqtt.password}")
    private String password;

    @Value("${mqtt.inBoundClientId}")
    private String inClientId;

    @Value("${mqtt.outBoundClientId}")
    private String outClientId;

    @Value("${mqtt.completionTimeout}")
    private Integer completionTimeout = 500000;

    @Value("${mqtt.defaultTopic}")
    private String defaultTopic;

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{url});
        if (StringUtils.hasText(username)) {
            options.setUserName(username);
        }
        if (StringUtils.hasText(password)) {
            options.setPassword(password.toCharArray());
        }
        options.setCleanSession(false);
        options.setKeepAliveInterval(24 * 60);
        options.setAutomaticReconnect(true);
        factory.setConnectionOptions(options);
        return factory;
    }

    private String[] getSubscribedTopics() {
        List<String> topicList = Arrays.asList(topics.split(",")).stream().filter(item -> item.trim().length() > 0).map(item -> item.trim()).collect(Collectors.toList());
        String[] topicArray;
        if (topicList.size() == 0) {
            topicArray = new String[1];
            if (StringUtils.hasText(defaultTopic)) {
                topicArray[0] = defaultTopic.split(",")[0];
            } else {
                topicArray[0] = "rhh/test";
            }
        } else {
            topicArray = new String[topicList.size()];
            topicList.toArray(topicArray);
        }
        return topicArray;
    }

    /**
     * 消息接收端
     */
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound() {
        String[] topicArray = getSubscribedTopics();
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(getInboundClientId(), mqttClientFactory(),
                        topicArray);
        adapter.setCompletionTimeout(completionTimeout);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(2);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return message -> {
            String index = String.valueOf(message.getHeaders().get(MqttHeaders.ID));
            String topic = String.valueOf(message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC));
            String qos = String.valueOf(message.getHeaders().get(MqttHeaders.RECEIVED_QOS));

            Long timestamp = (Long) message.getHeaders().get("timestamp");
            Date date = new Date();
            date.setTime(timestamp);
            String dateString = DateUtils.formatISO(date);
            Boolean retained = (Boolean) message.getHeaders().get(MqttHeaders.RECEIVED_RETAINED);
            String payload = message.getPayload().toString();
            System.out.printf("Terry -messageArrived> topic=%s, message=%s, 顺序号=%s, qos=%s, 时间:%s, 保留=%s\r\n", topic, payload, index, qos, dateString, retained);
        };
    }

    private String getInboundClientId() {
        return inClientId + "_" + System.currentTimeMillis();
    }
    /** 消息接收端 ****/

    /**
     * 消息发送
     ****/
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(outClientId, mqttClientFactory());
        // 如果设置成true，发送消息时将不会阻塞。
        messageHandler.setAsync(true);
        messageHandler.setDefaultQos(2);
        messageHandler.setAsyncEvents(true);
        messageHandler.setDefaultTopic("testTopic");
        return messageHandler;
    }

    @Component
    @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
    @ConditionalOnProperty(value = "mqtt.enabled", havingValue = "true")
    public interface MyGateway {

        /**
         * 定义重载方法，用于消息发送
         *
         * @param data
         */
        void sendToMqtt(String data);

        /**
         * 发送消息
         *
         * @param topic
         * @param payload
         */
        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);

        /**
         * 发送消息
         *
         * @param topic
         * @param payload
         * @param qos
         */
        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload, @Header(MqttHeaders.QOS) int qos);

    }

    /**
     * 发送消息
     */

    public void addTopic(String topic) {
        addTopic(topic, 2);
    }

    public void addTopic(String topic, int qos) {
    }

    public void removeTopc(String topic) {

    }
}